package dm.ass2.eval;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import dm.ass2.ClusteringDriver;
import dm.ass2.eval.io.ErrorWritable;

    public class EvalDriver extends Configured implements Tool {
    	
    	public static final String OUTPUT_REDUCE_DIR = "/evalTemp/";
    	
    	public static final String OUTPUT_FILENAME_NICE = "/eval";
		
		/**
		 * Generate the items code book, denoted V in the algorithm.
		 */	
		private boolean runProfilingJob(String inputPath) throws Exception {
			Job job = new Job(new Configuration());
			Configuration conf = job.getConfiguration();
			FileSystem fs = FileSystem.getLocal(conf);
		    Path tempOutputPath = new Path(ClusteringDriver.OUTPUT_DIR + OUTPUT_REDUCE_DIR);
		    if (fs.exists(tempOutputPath)) {
		    	fs.delete(tempOutputPath,true);			
			}	
			job.setJarByClass(EvalDriver.class);
			FileInputFormat.addInputPath(job, new Path(inputPath));
			//job.setOutputFormatClass(SequenceFileOutputFormat.class);
			//SequenceFileOutputFormat.setOutputPath(job, tempOutputPath);
			job.setOutputFormatClass(TextOutputFormat.class);
			FileOutputFormat.setOutputPath(job, tempOutputPath);

			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(Text.class);
			job.setMapOutputKeyClass(IntWritable.class);
			job.setMapOutputValueClass(ErrorWritable.class);
			job.setMapperClass(EvalMapper.class);
			job.setCombinerClass(EvalCombiner.class);
			job.setReducerClass(EvalReducer.class);
			return job.waitForCompletion(true);		
		}
		
		@Override
		public int run(String[] args) throws Exception {
			runProfilingJob(args[0]);
			
			return 0;

		}

		public static void main(String[] args) throws Exception {
			int ret = ToolRunner.run(new EvalDriver(), args);
			System.exit(ret);
		}
}
